package cn.cloud.service;

import cn.cloud.pojo.MQEntity;
import cn.cloud.utils.SerializeUtils;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;

/**
 * Created by Cloud on 2018/3/18 0018.
 */
public class ProducerServiceImpl implements ProducerService {
    private static Logger logger = LoggerFactory.getLogger(ProducerServiceImpl.class);

    private String namesrvAddr;

    private String producerGroup;

    private Boolean retryAnotherBrokerWhenNotStoreOK;

    private DefaultMQProducer producer;


    /**
     * spring 容器初始化所有属性后调用此方法
     */
    public void afterPropertiesSet() throws Exception {
        producer = new DefaultMQProducer();
        producer.setProducerGroup(this.producerGroup);
        producer.setNamesrvAddr(this.namesrvAddr);
        producer.setRetryAnotherBrokerWhenNotStoreOK(this.retryAnotherBrokerWhenNotStoreOK);
//        producer.setVipChannelEnabled(false);
        /*
         * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
         * 注意：切记不可以在每次发送消息时，都调用start方法
         */
        producer.start();
        logger.info("[{}:{}] start successd!", producerGroup, namesrvAddr);
    }

    /**
     * 销毁
     */
    public void destroy() throws Exception {
        if (producer != null) {
            logger.info("producer: [{}:{}] end ", producerGroup, namesrvAddr);
            producer.shutdown();
        }

    }


    public void send(String topic, String tags, MQEntity entity) {
        String keys = UUID.randomUUID().toString();
        entity.setKey(keys);
        entity.setTag(tags);
        logger.info("业务:{},tags:{},keys:{},entity:{}", topic, tags, entity.getKey(), entity);
        Message msg = new Message(topic, tags, keys,
                SerializeUtils.serialize(entity));
        try {
            producer.send(msg);
        } catch (Exception e) {
            logger.error(keys.concat(":发送消息失败"), e);
            throw new RuntimeException("发送消息失败", e);
        }
    }


    public void send(String topic, String tags, MQEntity entity, SendCallback sendCallback) {
        String keys = UUID.randomUUID().toString();
        entity.setKey(keys);
        entity.setTag(tags);
        logger.info("业务:{},tags:{},keys:{},entity:{}", topic, tags, keys, entity);
        Message msg = new Message(topic, tags, keys,
                SerializeUtils.serialize(entity));
        try {
            producer.send(msg, sendCallback);
        } catch (Exception e) {
            logger.error(keys.concat(":发送消息失败"), e);
            throw new RuntimeException("发送消息失败", e);
        }
    }


    public void sendOneway(String topic, String tags, MQEntity entity) {
        String keys = UUID.randomUUID().toString();
        entity.setKey(keys);
        entity.setTag(tags);
        logger.info("业务:{},tags:{},keys:{},entity:{}", topic, tags, keys, entity);
        Message msg = new Message(topic, tags, keys,
                SerializeUtils.serialize(entity));
        try {
            producer.sendOneway(msg);
        } catch (Exception e) {
            logger.error(keys.concat(":发送消息失败"), e);
            throw new RuntimeException("发送消息失败", e);
        }
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }


    public void setProducerGroup(String producerGroup) {
        this.producerGroup = producerGroup;
    }


    public void setProducer(DefaultMQProducer producer) {
        this.producer = producer;
    }


    public void setRetryAnotherBrokerWhenNotStoreOK(
            Boolean retryAnotherBrokerWhenNotStoreOK) {
        this.retryAnotherBrokerWhenNotStoreOK = retryAnotherBrokerWhenNotStoreOK;
    }
}
